package com.atguigu.Flink.datastream.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.KafkaSourceBuilder;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

public class Flink02_KafkaSource {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //kafka
        KafkaSource<String> kafkaSource =
                KafkaSource.<String>builder()
                        .setBootstrapServers("hadoop102:9092,hadoop103:9092,hadoop104:9092")
                        .setTopics("test")
                        .setGroupId("flink1")
                        //.setDeserializer()  // 设置 key和value的反序列化器
                        .setValueOnlyDeserializer(new SimpleStringSchema())  // 设置Value的反序列化器
                        //默认使用记录的offset进行消费， 如果需要重置offset，指定重置到LATEST
                        .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
                        //别的一些设置，如果没有提供直接的方法， 使用如下的方法进行设置
                        //.setProperty("enable.auto.commit" , "true" )
                        .setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG , "true")
                        //.setProperty("auto.commit.interval.ms" , "5000")
                        .setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG , "5000")

                        .build();

        DataStreamSource<String> ds = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka");
        ds.print();
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }
}
